
package com.phevos.common.util;


/**
 * poll thread. ����:�������/��ʱ�����ȹ��? ��������POLL,����POLL��ʵ�ֱ��̡߳� modify by zhaoch on
 * 20040609:��Sleepʱ,��ʱ��4�鿴�����б��Ƿ�仯.���仯,������Schedule;�������˯��
 */
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Vector;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PollThread extends Thread
{
    private static Logger logger = LoggerFactory.getLogger(PollThread.class.getName());

    private static final int TIMEOUT_SECONDS = 2; // ��ʱʱ��

    private static int currentThreadNumber = 0;

    // û������ʱ��Sleepʱ��,��λ:��
    private final static int SLEEPTIME_NOTASK = 5;

    // ��ʱ���˯ʱ,��Ҫ��ʱ��4��ʱ����(1 minute),��λ:��
    private final static int SLEEPTIME_INTERVAL = 30;

    private final static int SECOND = 1000;

    private boolean iAmKeepingRunning = true;

    private List taskList = new Vector(); // �������

    boolean used;

    // monitor for synchronized_taskList
    private Object monitor_taskList = new Object();

    // ��������Ƿ���仯�ı�־
    private boolean taskList_changed = false;

    private boolean suportTaskSort = false; // ��������Ƿ�֧������

    private boolean remainWhenFree = false; // ���������Ϊ��ʱ���Ƿ����̲߳��ͷ�

    protected String name = null;

    PollThread()
    {
        currentThreadNumber++;
        logger.info("Thread number of pool :" + currentThreadNumber);
    }

    /**
     * �������б������һ���µ�����
     * 
     * @param task
     */
    public void addTask(Task task)
    {
        synchronized (monitor_taskList)
        {
            taskList.add(task);
            if (suportTaskSort)
            {
                sortTasks();
            }
            taskList_changed = true;
            monitor_taskList.notifyAll();
        }
    }

    /**
     * �������б�����Ӷ���µ�����
     * 
     * @param task
     */
    public void addTasks(Task[] tasks)
    {
        synchronized (monitor_taskList)
        {
            for (int i = 0; i < tasks.length; i++)
            {
                taskList.add(tasks[i]);
            }
            if (suportTaskSort)
            {
                sortTasks();
            }
            taskList_changed = true;
            monitor_taskList.notifyAll();
        }
    }

    public Task getTask(String identifier)
    {
        synchronized (monitor_taskList)
        {
            int size = taskList.size();
            for (int i = 0; i < size; i++)
            {
                Task task = (Task)taskList.get(i);
                if (task.getIdentifier().equals(identifier))
                {
                    return task;
                }
            }
            return null;
        }
    }

    /**
     * �õ������Բ���ʼ��identifier������task
     * 
     * @param identifier
     * @return
     */
    public Task[] getTasks(String identifier)
    {
        synchronized (monitor_taskList)
        {
            List returnTasks = new ArrayList();
            int size = taskList.size();
            for (int i = 0; i < size; i++)
            {
                Task task = (Task)taskList.get(i);
                if (task.getIdentifier().startsWith(identifier))
                {
                    returnTasks.add(task);
                }
            }
            return (Task[])returnTasks.toArray(new Task[returnTasks.size()]);
        }
    }

    public void setSupportTaskSort(boolean b)
    {
        this.suportTaskSort = b;
    }

    public boolean freed()
    {
        return !this.used;
    }

    public void setRemainWhenFree(boolean b)
    {
        this.remainWhenFree = b;
    }

    public boolean isEmpty()
    {
        synchronized (monitor_taskList)
        {
            return taskList.isEmpty();
        }
    }

    public String toString()
    {
        return "[PollThread,name=" + name + ",used=" + this.used + ",tasks=" + taskList.size()
                + "]";
    }

    private void sortTasks()
    {
        try
        {
            java.util.Collections.sort(taskList);
        }
        catch (Exception ex)
        {
            logger.error("task list not support sort", ex);
        }
    }

    public void list()
    {
        StringBuffer sb = new StringBuffer();
        int size = taskList.size();
        sb.append("[Poll Thread, include ").append(size).append(" tasks:");
        for (int i = 0; i < size; i++)
        {
            sb.append(((Task)taskList.get(i)).getIdentifier()).append(",");
        }
        sb.append("]");
        // ###logger.debug(sb.toString());
    }

    /**
     * ΪHouseKeeeper�߳��ṩ�����б��еĳ�ʱ����,��ʱʱ��ʹ��TIMEOUT_SECONDS
     * 
     * @return
     * @author wangmh modify by zhaoch on 20040908:
     *         ����Task���������ʱ������(��:�������ʱ��>0)����TIMEOUTʱ��ʹ��������������ʱ��; ����(��:�������ʱ��<=0)��ʹ��TIMEOUT_SECONDS
     *         --end modify ����һ��note:һ�����������ܹ����жϣ����뱣֤�������ܹ���ӦThread.interrup������ ����Ĳ��ܱ�interrup�������:
     *         Socket��read/write����£�����д���ܱ����㣬���ҶԷ���ִ��close����
     *         Ϊ�˱�֤һ��Socket��������ֹ���������ʹ��socket.setSotimeout(<interval>)��
     */
    void interruptTimeOutTask()
    {
        if (taskList.isEmpty())
        {
            return;
        }

        // ��ʹ��ͬ��taskList�ķ�������Ϊ��Taskִ�в�����ʱtaskList��ִ���̱߳���
        int size = taskList.size();
        long timeoutMills = 0;
        for (int i = 0; i < size; i++)
        {
            Task task = (Task)taskList.get(i);
            if (task.maxExecuteTime <= 0)
            {
                timeoutMills = TIMEOUT_SECONDS * SECOND;
            }
            else
            {
                timeoutMills = task.maxExecuteTime * SECOND;
            }
            if (task.startTime > 0 && System.currentTimeMillis() - task.startTime >= timeoutMills)
            {
                logger.warn("Task " + task.getIdentifier() + " has run "
                        + (System.currentTimeMillis() - task.startTime) / SECOND
                        + " seconds , It was time out and interrupted.");
                this.interrupt();
                return;
            }
        }
    }

    /**
     * ��� identifier ��������������ɾ�������б��������Բ���ʼidentifier ������ Ȼ���������б��е�������������Դ�����÷���ı�ʱ��ˢ��������
     * 
     * @param groupIden ��־һ��������ĳ��Դ�Ĺ�������
     * @param taskGroup
     */
    public void setTasks(String groupIden, Task[] tasks)
    {
        synchronized (monitor_taskList)
        {
            for (int i = 0; i < taskList.size();)
            {
                Task task = (Task)taskList.get(i);
                if (task.getIdentifier().startsWith(groupIden))
                {
                    task.clear();
                    taskList.remove(task);
                    taskList_changed = true;
                }
                else
                {
                    i++;
                }
            }
            if (tasks != null && tasks.length > 0)
            {
                for (int i = 0; i < tasks.length; i++)
                {
                    taskList.add(tasks[i]);
                }
                taskList_changed = true;
            }
            if (taskList_changed)
            {
                monitor_taskList.notifyAll();
            }
            if (taskList.size() > 0 && suportTaskSort)
            {
                sortTasks();
            }
            if (taskList.isEmpty() && !remainWhenFree)
            { // ������пգ��ͷ��߳�
                PollThreadPool.free(this);
            }
        }
    }

    /**
     * �������б���ɾ��һ���������������пգ����ͷ��̻߳س���
     * 
     * @param task
     */
    public void removeTask(Task task)
    {
        synchronized (monitor_taskList)
        {
            task.clear();
            taskList.remove(task);
            if (suportTaskSort)
            {
                sortTasks();
            }

            if (taskList.isEmpty() && !remainWhenFree)
            { // ������пգ��ͷ��߳�
                PollThreadPool.free(this);
            }
        }
    }

    /**
     * ɾ�������б������Բ���ʼidentifier ��һ������
     * 
     * @param identifier
     */
    public Task removeTask(String identifier)
    {
        Task ret = null;
        synchronized (monitor_taskList)
        {
            int size = taskList.size();
            for (int i = 0; i < size; i++)
            {
                Task task = (Task)taskList.get(i);
                if (task.getIdentifier().equals(identifier))
                {
                    task.clear();
                    taskList.remove(task);
                    ret = task;
                    break;
                }
            }
            if (taskList.isEmpty() && !remainWhenFree)
            { // ������пգ��ͷ��߳�
                PollThreadPool.free(this);
            }
            return ret;
        }
    }

    /**
     * ɾ�������б��������Բ���ʼidentifier ������
     * 
     * @param identifier
     */
    public void removeTasks(String identifier)
    {
        synchronized (monitor_taskList)
        {
            for (int i = 0; i < taskList.size();)
            {
                Task task = (Task)taskList.get(i);
                if (task.getIdentifier().startsWith(identifier))
                {
                    task.clear();
                    taskList.remove(task);
                }
                else
                {
                    i++;
                }
            }
            if (taskList.isEmpty() && !remainWhenFree)
            { // ������пգ��ͷ��߳�
                PollThreadPool.free(this);
            }
            else if (suportTaskSort)
            {
                sortTasks();
            }
        }
    }

    /**
     * �������б���ɾ����������
     * 
     * @param task
     */
    public void removeAllTasks()
    {
        synchronized (monitor_taskList)
        {
            int size = taskList.size();
            for (int i = 0; i < size; i++)
            {
                Task task = (Task)taskList.get(i);
                task.clear();
            }
            taskList.clear();
        }
        // ������пգ��ͷ��߳�
        if (!remainWhenFree)
        {
            PollThreadPool.free(this);
        }
    }

    public void run()
    {
        boolean hasTask = false;
        while (iAmKeepingRunning)
        {
            synchronized (monitor_taskList)
            {
                if (taskList.size() > 0)
                {
                    hasTask = true;
                }
                else
                {
                    hasTask = false;
                }
            }// end synchronized
            if (hasTask)
            {
                beforeSleep();
                afterSleep();
            }
            else
            {
                try
                {
                    Thread.sleep(SLEEPTIME_NOTASK * SECOND);
                }
                catch (Exception e)
                {
                    e.getMessage();
                }
            }
        }
    }

    /**
     * �رձ��߳�
     */
    public void shutdown()
    {
        removeAllTasks(); // ɾ���������񣬲��ر���Դ
        this.iAmKeepingRunning = false;
    }

    public Task[] allTasks()
    {
        synchronized (monitor_taskList)
        {
            return (Task[])taskList.toArray(new Task[taskList.size()]);
        }
    }

    public void setRefName(String s)
    {
        name = s;
    }

    public String getRefName()
    {
        return name;
    }

    public void dump()
    {
        Task[] tasks = (Task[])taskList.toArray(new Task[taskList.size()]);
        StringBuffer sb = new StringBuffer();
        sb.append("Poll Thread, include ").append(tasks.length).append(" tasks:");
        // ###logger.debug(sb.toString());
        for (int i = 0; i < tasks.length; i++)
        {
            tasks[i].dump();
        }
    }

    // /////////////////////PRIVATE METHODS/////////////////////////
    /* sleepǰ���� */
    private void beforeSleep()
    {
        int minSleepTime = Integer.MAX_VALUE;
        int tmpSleepTime;
        Task task = null;
        Iterator ir = null;
        synchronized (monitor_taskList)
        {
            ir = taskList.iterator();
            // get min sleep time
            while (ir.hasNext())
            {
                task = (Task)ir.next();
                tmpSleepTime = task.nextTime;
                if (tmpSleepTime <= minSleepTime)
                    minSleepTime = tmpSleepTime;
            }
        }// end synchronized

        int leftSleepTime = minSleepTime;
        while (leftSleepTime > 0)
        {
            if (leftSleepTime >= SLEEPTIME_INTERVAL)
            {
                tmpSleepTime = SLEEPTIME_INTERVAL;
            }
            else
            {
                tmpSleepTime = leftSleepTime;
            }
            leftSleepTime -= tmpSleepTime;
            if (tmpSleepTime > 0)
            {
                try
                {
                    Thread.sleep(tmpSleepTime * SECOND);
                }
                catch (Exception e)
                {
                    // ###logger.debug("thread interrupted when sleeping in
                    // beforeSleep");
                }
            }
            // ����,��4,Ҫ������������Ƿ���仯.���������з���仯������Schedule
            synchronized (monitor_taskList)
            {
                ir = taskList.iterator();
                while (ir.hasNext())
                {
                    task = (Task)ir.next();
                    task.nextTime -= tmpSleepTime;
                }
                if (taskList_changed)
                {
                    taskList_changed = false;
                    break;
                }
            }// end synchronized
        }
    }

    /* sleep����� */
    private void afterSleep()
    {
        synchronized (monitor_taskList)
        {
            Task task = null;
            // ���û��������ȴ�
            while (taskList.isEmpty())
            {
                try
                {
                    monitor_taskList.wait();
                }
                catch (InterruptedException ie)
                {
                    // ###logger.debug("thread interupted when wait for
                    // wait()");
                }
            }
            // get min sleep time
            for (int i = 0; i < taskList.size(); i++)
            {
                task = (Task)taskList.get(i);
                if (task == null)
                {
                    logger.warn("in taskList: task is null");
                    continue;
                }
                if (task.nextTime <= 0)
                {
                    try
                    {
                        executeTask(task);
                        if (task.over())
                        { // ����������ɾ������
                            removeTask(task);
                        }
                    }
                    catch (Throwable t)
                    {
                        logger.error("ִ��poll������", t);
                    }
                    task.nextTime += task.interval;
                }
            }
        }
    }

    private static void executeTask(Task task) throws Exception
    {
        task.startTime = System.currentTimeMillis();
        // �������̳���ע�ὫҪִ�е�����
        if (task.isbestrow())
        {
            TaskThreadPool.getInstance().registTask(task);
        }
        else
        {
            task.run();
        }
        task.startTime = 0;
    }

}
